package com.bff.gaia.mix.examples.xjoin.v3;/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.bff.gaia.api.common.functions.MapFunction;
import com.bff.gaia.api.common.typeinfo.TypeInformation;
import com.bff.gaia.api.java.ExecutionEnvironment;
import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.mix.api.TimeCharacteristic;
import com.bff.gaia.mix.api.datastream.DataMixStream;
import com.bff.gaia.mix.api.environment.MixStreamExecutionEnvironment;
import com.bff.gaia.mix.api.xjoin.core.enums.ECacheType;
import com.bff.gaia.mix.api.xjoin.core.enums.EJoinType;
import com.bff.gaia.mix.api.xjoin.client.rdb.table.RdbSideTableInfo;
import com.bff.gaia.mix.api.xjoin.client.rdb.parser.RdbSideTableInfoBuilders;
import com.bff.gaia.mix.examples.xjoin.benchmarkDataSource;
import com.bff.gaia.types.Row;


/**
 * Project Name: gaia-parent
 * Description:
 * Data: 2019/7/24 14:25
 * Company: NEU
 *
 * @author tiger
 * @version v1.0
 */
public class xjoinExampleV3 {
	public static void main(String[] args) throws Exception {
		MixStreamExecutionEnvironment env = MixStreamExecutionEnvironment.getExecutionEnvironment();
		env.setMixStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
		final ExecutionEnvironment eenv = ExecutionEnvironment.getExecutionEnvironment();

		final String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/employees?characterEncoding=UTF-8&useSSL=false&serverTimezone=Hongkong&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true";
		final String jdbcUser = "gaia";
		final String jdbcPassword = "labmysqlgaia";

		RdbSideTableInfo dept_empTable = RdbSideTableInfoBuilders.buildMysqlTableInfo()
			.setTableName("dept_emp")
			.setUrl(jdbcUrl)
			.setUserName(jdbcUser)
			.setPassword(jdbcPassword)
			.addfieldInfos(new String[]{"emp_no int", "dept_no string", "from_date date", "to_date date"})
			.finish();

		RdbSideTableInfo deptTable = RdbSideTableInfoBuilders.buildMysqlTableInfo()
			.setTableName("departments")
			.setUrl(jdbcUrl)
			.setUserName(jdbcUser)
			.setPassword(jdbcPassword)
			.addfieldInfos(new String[]{"dept_no string", "dept_name string"})
			.finish();

		RdbSideTableInfo empTable = RdbSideTableInfoBuilders.buildMysqlTableInfo()
			.setTableName("employees")
			.setUrl(jdbcUrl)
			.setUserName(jdbcUser)
			.setPassword(jdbcPassword)
			.addfieldInfos(new String[]{
				"emp_no int",
				"birth_date date",
				"first_name string",
				"last_name string",
				"gender string",
				"hire_date date"})
			.finish();

		int num = 10;
		if (args.length == 1) {
			num = Integer.valueOf(args[0]);
		}
		DataMixStream<Integer> empNoStream = env.addSource(new benchmarkDataSource.SimpleSource(num));
		RowTypeInfo emp=new RowTypeInfo(TypeInformation.of(Integer.TYPE));
		DataMixStream<Row> empNoTupleStream = empNoStream.map(new MapFunction<Integer, Row>() {
			@Override
			public Row map(Integer value) throws Exception {
				Row row = new Row(1);
				row.setField(0, value);
				return row;
			}
		}).returns(emp);

//		XjoinMixStreamV3 x=new XjoinMixStreamV3(empNoTupleStream,dept_empTable);
		//连接 员工部门表
		DataMixStream dept_empStream=empNoTupleStream
			.xjoin(dept_empTable)
			.projectLeftKey(0)
			.projectRightKey(0)
			.projectRightField(0,1)
			.setCacheType(ECacheType.ALL)
			.apply(EJoinType.EQUAL)
			.setParallelism(1);
		dept_empStream.print();
//		System.out.println(dept_empStream.getType());
////		扩展员工信息
//		DataMixStream empExStream=dept_empStream
//			.xjoin(empTable)
//			.projectLeftKey(0)
//			.projectRightKey(0)
//			.projectRightField(0,2,3)
//			.setCacheType(ECacheType.LRU)
//			.apply(EJoinType.EQUAL)
//			.setParallelism(1);
////		empExStream.print();
//
//		//扩展部门信息
//		DataMixStream empExAndDeptExStream=empExStream
//			.xjoin(deptTable)
//			.projectLeftKey(1)
////			.projectLeftField(0,1)
//			.projectRightKey(0)
//			.setCacheType(ECacheType.LRU)
//			.apply(EJoinType.EQUAL)
//			.setParallelism(1);
//
//		empExAndDeptExStream.print();
		env.execute();
	}
}